/**
 * 服务端 SSE (Server-Sent Events) 管理器
 * 提供统一的 SSE 连接管理、消息分发和会话维护功能
 */

const { getLogger } = require('../logger');
const logger = getLogger('infra.sseManager');

class SSEManager {
    static instance = null;
    constructor() {
        this.connections = new Map();
        this.userConnections = new Map();
        this.connectionUsers = new Map();
        this.options = {
            heartbeatInterval: 30000,
            defaultEventType: 'message',
            connectionTimeout: 120000
        };
        this.heartbeatTimer = null;
        this.startHeartbeat();
        logger.info('SSE 管理器已初始化');
    }
    static getInstance() {
        if (!SSEManager.instance) {
            SSEManager.instance = new SSEManager();
        }
        return SSEManager.instance;
    }
    configure(options) {
        this.options = { ...this.options, ...options };
    }
    createConnection(req, res, userId = null) {
        const connectionId = `conn_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
        res.writeHead(200, {
            'Content-Type': 'text/event-stream',
            'Cache-Control': 'no-cache',
            'Connection': 'keep-alive',
            'X-Accel-Buffering': 'no'
        });
        req.socket.setTimeout(0);
        const connection = { req, res, createdAt: Date.now(), lastActive: Date.now(), userId, status: 'active' };
        const clientId = (req.query && req.query.clientId) || req.headers['x-client-id'] || null;
        if (clientId) connection.clientId = clientId;
        this.connections.set(connectionId, connection);
        if (userId) this._associateUserWithConnection(userId, connectionId);
        try {
            const initialMessage = `event: open\ndata: ${JSON.stringify({ connectionId, clientId: connection.clientId || null })}\n\n`;
            res.write(initialMessage);
            logger.info('已创建新的 SSE 连接', { connectionId, userId });
        } catch (error) {
            logger.error('发送初始连接消息失败', { connectionId, userId, error });
            this.closeConnection(connectionId);
            throw error;
        }
        req.on('close', () => {
            if (this.connections.has(connectionId)) {
                connection.status = 'closed';
                this.closeConnection(connectionId);
            }
        });
        req.on('error', (error) => {
            logger.error('SSE 连接错误', { connectionId, error });
            if (this.connections.has(connectionId)) {
                connection.status = 'error';
                this.closeConnection(connectionId);
            }
        });
        return connectionId;
    }
    sendMessage(connectionId, message) {
        const connection = this.connections.get(connectionId);
        if (!connection) {
            logger.warn('尝试发送消息到不存在的连接', { connectionId });
            return false;
        }
        if (connection.status !== 'active') {
            logger.warn('尝试发送消息到非活跃连接', { connectionId, status: connection.status });
            return false;
        }
        try {
            const eventType = message.type || this.options.defaultEventType;
            const payload = typeof message.data === 'string' ? message.data : JSON.stringify(message.data);
            const sseMessage = `event: ${eventType}\ndata: ${payload}\n\n`;
            if (!connection.res.writable) {
                logger.warn('响应流不可写', { connectionId });
                this.closeConnection(connectionId);
                return false;
            }
            connection.res.write(sseMessage);
            connection.lastActive = Date.now();
            if (connection.res.flushHeaders && typeof connection.res.flushHeaders === 'function') {
                connection.res.flushHeaders();
            }
            return true;
        } catch (error) {
            logger.error('发送消息到连接失败', { connectionId, error: error });
            if (error.code !== 'ECONNRESET' && error.code !== 'EPIPE') {
                this.closeConnection(connectionId);
            }
            return false;
        }
    }
    sendToUser(userId, message) {
        const userConns = this.userConnections.get(userId) || new Set();
        let sentCount = 0;
        userConns.forEach(connectionId => {
            if (this.sendMessage(connectionId, message)) {
                sentCount++;
            }
        });
        if (sentCount === 0 && userConns.size > 0) {
            logger.warn('向用户的所有连接发送消息失败', { userId, connectionCount: userConns.size });
        }
        return sentCount;
    }
    broadcast(message, filterFn = null) {
        let sentCount = 0;
        this.connections.forEach((connection, connectionId) => {
            if (!filterFn || filterFn(connection, connectionId)) {
                if (this.sendMessage(connectionId, message)) {
                    sentCount++;
                }
            }
        });
        logger.info('广播消息完成', { successCount: sentCount, totalCount: this.connections.size });
        return sentCount;
    }
    checkHeartbeats() {
        const now = Date.now();
        const timeoutThreshold = this.options.connectionTimeout;
        let inactiveCount = 0;
        let heartbeatFailedCount = 0;
        this.connections.forEach((connection, connectionId) => {
            try {
                if (!this.sendMessage(connectionId, { type: 'heartbeat', data: { timestamp: now } })) {
                    heartbeatFailedCount++;
                }
                if (now - connection.lastActive > timeoutThreshold) {
                    logger.info('SSE 连接超时', { connectionId, lastActive: new Date(connection.lastActive).toISOString(), duration: now - connection.lastActive });
                    this.closeConnection(connectionId);
                    inactiveCount++;
                }
            } catch (error) {
                logger.error('心跳检查过程中出错', { connectionId, error: error });
                try {
                    this.closeConnection(connectionId);
                } catch (closeError) {
                    logger.error('关闭出错连接失败', { connectionId, error: closeError.message });
                }
            }
        });
        if (inactiveCount > 0 || heartbeatFailedCount > 0) {
            logger.info('SSE 心跳检查完成', { closedInactiveConnections: inactiveCount, heartbeatFailedConnections: heartbeatFailedCount, totalConnections: this.connections.size });
        }
    }
    cleanupInactiveConnections() {
        const now = Date.now();
        const timeoutThreshold = this.options.connectionTimeout;
        const connectionsToClose = [];
        this.connections.forEach((connection, connectionId) => {
            if (now - connection.lastActive > timeoutThreshold) {
                connectionsToClose.push(connectionId);
            }
        });
        connectionsToClose.forEach(connectionId => {
            this.closeConnection(connectionId);
        });
        logger.info('SSE 清理完成', { removedConnections: connectionsToClose.length });
        return connectionsToClose.length;
    }
    getStats() {
        const uniqueUsers = new Set();
        this.userConnections.forEach((_, userId) => {
            uniqueUsers.add(userId);
        });
        return { totalConnections: this.connections.size, totalUsers: uniqueUsers.size, connectionsByUser: Object.fromEntries(Array.from(this.userConnections.entries()).map(([userId, connSet]) => [userId, connSet.size])) };
    }
    closeConnection(connectionId) {
        const connection = this.connections.get(connectionId);
        if (!connection) {
            logger.warn('尝试关闭不存在的连接', { connectionId });
            return false;
        }
        if (connection.status === 'closing' || connection.status === 'closed') {
            logger.warn('尝试关闭已关闭的连接', { connectionId, status: connection.status });
            this._cleanupConnectionMappings(connectionId);
            return false;
        }
        connection.status = 'closing';
        const userId = this.connectionUsers.get(connectionId);
        try {
            try {
                const closeMessage = `event: close\ndata: ${JSON.stringify({ reason: 'server_shutdown' })}\n\n`;
                if (connection.res.writable) {
                    connection.res.write(closeMessage);
                }
            } catch (msgError) {
                logger.warn('发送关闭事件失败，但继续关闭连接', { connectionId, error: msgError.message });
            }
            if (connection.res && typeof connection.res.end === 'function') {
                try { connection.res.end(); } catch (endError) { logger.warn('响应流关闭时出错', { connectionId, error: endError.message }); }
            }
            this._cleanupConnectionMappings(connectionId);
            connection.status = 'closed';
            this.connections.delete(connectionId);
            logger.info('SSE 连接已成功关闭', { connectionId, userId, duration: Date.now() - connection.createdAt });
            return true;
        } catch (error) {
            logger.error('关闭 SSE 连接过程中出错', { connectionId, error: error });
            try { this._cleanupConnectionMappings(connectionId); this.connections.delete(connectionId); } catch (cleanupError) { logger.error('清理连接映射失败', { connectionId, error: cleanupError.message }); }
            return false;
        }
    }
    closeUserConnections(userId) {
        const userConns = this.userConnections.get(userId) || new Set();
        let closedCount = 0;
        userConns.forEach(connectionId => { if (this.closeConnection(connectionId)) { closedCount++; } });
        logger.info('已关闭用户的连接', { userId, closedCount });
        return closedCount;
    }
    getUserConnectionCount(userId) {
        const userConns = this.userConnections.get(userId);
        return userConns ? userConns.size : 0;
    }
    getTotalConnectionCount() { return this.connections.size; }
    getConnectionInfo(connectionId) {
        const connection = this.connections.get(connectionId);
        if (!connection) return null;
        return { connectionId, userId: connection.userId, createdAt: connection.createdAt, lastActive: connection.lastActive, duration: Date.now() - connection.createdAt };
    }
    updateConnectionUser(connectionId, userId) {
        const connection = this.connections.get(connectionId);
        if (!connection) return false;
        const oldUserId = this.connectionUsers.get(connectionId);
        if (oldUserId) this._disassociateUserWithConnection(oldUserId, connectionId);
        connection.userId = userId;
        if (userId) this._associateUserWithConnection(userId, connectionId);
        logger.info('已更新连接的用户信息', { connectionId, userId });
        return true;
    }
    startHeartbeat() {
        if (this.heartbeatTimer) { clearInterval(this.heartbeatTimer); }
        this.heartbeatTimer = setInterval(() => { this.checkHeartbeats(); }, this.options.heartbeatInterval);
        logger.info('SSE 心跳机制已启动', { interval: this.options.heartbeatInterval });
    }
    stopHeartbeat() { if (this.heartbeatTimer) { clearInterval(this.heartbeatTimer); this.heartbeatTimer = null; logger.info('SSE 心跳机制已停止'); } }
    _associateUserWithConnection(userId, connectionId) { if (!this.userConnections.has(userId)) { this.userConnections.set(userId, new Set()); } this.userConnections.get(userId).add(connectionId); this.connectionUsers.set(connectionId, userId); }
    _disassociateUserWithConnection(userId, connectionId) { const userConns = this.userConnections.get(userId); if (userConns) { userConns.delete(connectionId); if (userConns.size === 0) { this.userConnections.delete(userId); } } this.connectionUsers.delete(connectionId); }
    _cleanupConnectionMappings(connectionId) { const userId = this.connectionUsers.get(connectionId); if (userId) { this._disassociateUserWithConnection(userId, connectionId); } this.connectionUsers.delete(connectionId); }
    shutdown() { logger.info('正在关闭所有 SSE 连接...'); this.stopHeartbeat(); const connectionIds = Array.from(this.connections.keys()); connectionIds.forEach(connectionId => { this.closeConnection(connectionId); }); logger.info('已关闭所有 SSE 连接', { closedCount: connectionIds.length }); }
}

module.exports = SSEManager;